package org.example

import com.alibaba.fastjson.JSON
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.parsing.json.JSONObject

/**
 * @author yangzhen14
 * @create 2021/4/20 12:50
 */
object stream {
  def main(Args:Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "C:\\\\Users\\\\yangzhen14\\\\hadoop-2.7.1")
//    val conf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")
//    //设置 10秒读取一次
//    val ssc = new StreamingContext(conf, Seconds(10))
////    ssc.textFile("/user/data/201908/0[1-9]/*")
//    //监听指定的文件夹
//    //它不会读取原本就已经存在于该文件夹里的文本，只会读取在监听期间，传进文件夹的文本，而且该文本还有要求，必须是它最后一次更改并且保存的操作，是在监听开始的那一刻之后的。
//    //    其实意思就是，如果要向被监听的文件夹里传一个文本，你就要在监听开始之后，先打开这个文本，随便输入几个空格，或者回车，或者其他不影响文本内容的操作，然后保存，最后再传进文件夹里，这样它才能检测到这个被传进来的文本。（估计它这个用意是只监听被更改过的文本吧）
//
//    val lines = ssc.textFileStream("C:\\tmp\\")
//    lines.print()
//    val filterLine_yz = lines.filter(line => line.contains("yz")).map(x=>"yz")
//    val filterLine_hello = lines.filter(line => line.contains("hello")).map(x=>"hello")
//    val wordCounts_yz = filterLine_yz.map((_, 1)).reduceByKey(_+_)
//    val wordCounts_hello = filterLine_hello.map((_, 1)).reduceByKey(_+_)
//    println("wordCounts_hello.count():")
//    wordCounts_hello.print()
//    println("wordCounts_yz.count():")
//    wordCounts_yz.print()
//    lines.foreachRDD( rdd =>{
//      rdd.foreachPartition(partitionOfRecords =>{
//        val connect = scalaConnectPool.getConnection
//        connect.setAutoCommit(false)
//        val stmt = connect.createStatement()
//        partitionOfRecords.foreach(record =>{
//          println("操作数据库了 start")
//          println(record(0))
//          stmt.addBatch("insert into test (insert_time,search_count) values (now(),'"+record(0)+"')")
//          println("操作数据库了 end ")
//        })
//        stmt.executeBatch()
//        connect.commit()
//      }
//      )
//    }
//    )
//    println("111111111111111111111111111111111111111111111111111111")
//    println(lines.count())
//
//    ssc.start()
//    ssc.awaitTermination()


//    val conf = new SparkConf().setAppName("HelloWorld").setMaster("spark://10.122.11.116:7077")




    val conf = new SparkConf().setAppName("HelloWorld").setMaster("local[*]")

    val sc = new SparkContext(conf)

    val lines = sc.textFile("C:\\tmp\\*.txt")

    println(lines.count())
//    lines.collect().foreach(
//      line=>{
//
//        val array = line.split(" ")
//        println(array.size)
//        //正常顺序遍历
//        for(i <- 0 until array.length){
//          println(array(i))
//        }
//        array.foreach(i=>{
//
//        })
//        println("数据库操作一行："+line)
//      }
//    )
    val filterLine_yz = lines.filter(line => line.contains("yz")).map(x=>"yz")

    val filterLine_hello = lines.filter(line => line.contains("hello")).map(x=>"hello")
        val wordCounts_yz = filterLine_yz.map((_, 1)).reduceByKey(_+_)
        val wordCounts_hello = filterLine_hello.map((_, 1)).reduceByKey(_+_)

//        println("wordCounts_hello.count():"+wordCounts_hello.collect()(1))
        wordCounts_hello.foreach((x)=>{
          println(x)

        })
    wordCounts_yz.foreach((x)=>{
      println(x._1)

    })




  }
}
